import pandas as pd
from es_pandas import es_pandas

from air_web.config.config import config
from air_web.data_platform import init_db
from air_web.dw.data_mapping import ConsType


class MergeCons:
    """生成merge_cons"""

    def __init__(self, *args, **kwargs):
        self.sql_engine = init_db()
        self.save_table = config.get("MERGE_CONS", "merge_cons")

    def create_cons_merge(self):
        has_df = self.sql_engine.query("""SELECT * FROM merge_cons""")
        if len(has_df) == 0:
            sql = f"""select distinct 
                           t1.on5, t1.shi, t3.ad_org_name ad_shi, 
                           t1.on7, t1.xian, t2.ad_org_name ad_xian, 
                           t6.org_no province_no, t6.org_name province_name,
                           t1.type_id, t1.type_code, 
                           t1.pare_type_id, t1.pare_type_code, 
                           t1.type_code_sort,
                           t5.type_code as sort_name,
                           t5.p_type_id highest_type_id
                           from 
                            (select distinct on5, shi, on7, xian, type_id, type_code, 
                                    pare_type_id, pare_type_code, type_code_sort 
                             from c_cons 
                             where cons_type in ({ConsType.ORDINARY_USER}, {ConsType.LINE_USER})) t1
                           left join real_org_no t2 on t1.on7 = t2.org_no
                           left join real_org_no t3 on t1.on5 = t3.org_no
                           left join type_map t5 on type_level = 0 and t1.type_code_sort = t5.type_id
                           left join real_org_no t6 on t6.org_level = 0"""
            merge_df = self.sql_engine.query(sql)
            merge_df["type_id"] = merge_df["type_id"].astype("int64")
            self.sql_engine.update_df_by_id(merge_df, self.save_table)
            print(f"写入mysql:{self.save_table},数据条数:{len(merge_df)}")

    def main(self):
        print(f"正在生成:{self.save_table}")
        self.sql_engine.truncate_table(self.save_table)
        self.create_cons_merge()


class CCons:
    """生成c_cons"""

    def __init__(self, *args, **kwargs):
        self.ep = es_pandas(config["ES_HOST"])
        self.sql_engine = init_db()
        self.save_table = config.get("C_CONS", "c_cons")

    def get_cons_df(self):
        df = self.ep.to_pandas(config["C_CONS_IDX"], show_progress=False)
        df["is_cal"] = df["if_cal"]
        if config["c_cons_filter_is_cal"]:
            df = df[df["is_cal"] == 1]
        res_df = df[
            [
                "on5",
                "shi",
                "on7",
                "xian",
                "type_id",
                "type_code",
                "type_code_sort",
                "cons_no",
                "cons_name",
                "cons_type",
                "pare_type_id",
                "pare_type_code",
                "org_no",
                "org_name",
                "is_cal",
                "run_cap",
                "contract_cap"
            ]
        ].drop_duplicates()
        self.sql_engine.update_df_by_id(res_df, self.save_table)
        print(f"写入mysql:{self.save_table},数据条数:{len(res_df)}")

    def main(self):
        print(f"正在生成:{self.save_table}")
        self.sql_engine.truncate_table(self.save_table)
        self.get_cons_df()


class AreaTypeMap:
    """生成area_type_map"""

    def __init__(self, *args, **kwargs):
        self.sql_engine = init_db()
        self.save_table = config.get("AREA_TYPE_MAP", "area_type_map")
        self.res_df = pd.DataFrame()

    def save_data_to_table(self):
        self.sql_engine.update_df_by_id(self.res_df, self.save_table)
        print(f"写入mysql:{self.save_table},数据条数:{len(self.res_df)}")

    def proc_area_type_map(self):
        org_sql_list = [
            "on7 as org_no, xian as org_name, 2 org_level",
            "on5 as org_no, shi as org_name, 1 org_level",
        ]
        type_sql_list = [
            "type_id, type_code, pare_type_id as p_type_id, 2 type_level",
            "pare_type_id as type_id, pare_type_code as type_code, type_code_sort as p_type_id,1 type_level",
        ]

        for org_sql in org_sql_list:
            for type_sql in type_sql_list:
                sql = """select distinct {org_sql},{type_sql}
                         from c_cons 
                         where cons_type = {cons_type}
                      """.format(
                    org_sql=org_sql,
                    type_sql=type_sql,
                    cons_type=ConsType.ORDINARY_USER,
                )
                self.res_df = self.sql_engine.query(sql)
                self.save_data_to_table()

            sql = """select distinct {org_sql},
                            type_code_sort as type_id,
                            tm.type_code,
                            tm.p_type_id,
                            0 type_level
                     from (select distinct {org_sql2}, type_code_sort from c_cons 
                     where cons_type = {cons_type}) cc
                     left join type_map tm on tm.type_id = cc.type_code_sort
                  """.format(
                org_sql=org_sql,
                org_sql2=",".join(
                    [i.split(" as")[0] for i in org_sql.split(",")]
                ),
                cons_type=ConsType.ORDINARY_USER,
            )
            self.res_df = self.sql_engine.query(sql)
            self.save_data_to_table()

        for type_sql in type_sql_list:
            sql = """select distinct ro.org_no, ro.org_name, 0 org_level,
                            {type_sql}
                     from (select distinct {type_sql2} from c_cons 
                     where cons_type = {cons_type}) cc
                     left join real_org_no ro on ro.org_level = 0
                  """.format(
                type_sql=type_sql,
                type_sql2=",".join(
                    [i.split(" as")[0] for i in type_sql.split(",")]
                ),
                cons_type=ConsType.ORDINARY_USER,
            )
            self.res_df = self.sql_engine.query(sql)
            self.save_data_to_table()

        sql = """select distinct ro.org_no, ro.org_name, 0 org_level,
                        type_code_sort as type_id,
                        tm.type_code,
                        tm.p_type_id,
                        0 type_level
                 from (select distinct type_code_sort from c_cons 
                 where cons_type = {cons_type}) cc
                 left join type_map tm on tm.type_id = cc.type_code_sort
                 left join real_org_no ro on ro.org_level = 0
              """.format(
            cons_type=ConsType.ORDINARY_USER
        )
        self.res_df = self.sql_engine.query(sql)
        self.save_data_to_table()

    def main(self):
        print(f"正在生成:{self.save_table}")
        self.sql_engine.truncate_table(self.save_table)
        self.proc_area_type_map()


class InitJudge:
    def __init__(self, *args, **kwargs):
        pass

    def main(self):
        sql_engine = init_db()
        sql_tm = "select * from type_map limit 1"
        tm_df = sql_engine.query(sql_tm)
        if tm_df.empty:
            raise PermissionError("type_map为空")

        sql_tml = "select * from type_map where type_level = 0 limit 1"
        tml_df = sql_engine.query(sql_tml)
        if tml_df.empty:
            raise PermissionError("type_map的type_level不是从0开始")

        sql_ron = "select * from real_org_no limit 1"
        ron_df = sql_engine.query(sql_ron)
        if ron_df.empty:
            raise PermissionError("real_org_no为空")

        sql_ronl = "select * from real_org_no where org_level = 0 limit 1"
        ronl_df = sql_engine.query(sql_ronl)
        if ronl_df.empty:
            raise PermissionError("real_org_no的org_level不是从0开始")


if __name__ == "__main__":
    InitJudge().main()
    CCons().main()
    MergeCons().main()
    AreaTypeMap().main()
